package cn.jslancer.spider;

import cn.jslancer.domain.Article;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.*;
import java.util.stream.Collectors;

/**
 * @author jslancer
 * @date 2017/11/5
 */
public class Scheduler {

    private UrlManager urlManager;
    private Downloader downloader;
    private List<DataPipeline> dataPipelines;
    private Spider spider;
    private int size = 10;

    //    ThreadPoolExecutor
    private ExecutorService executorService = Executors.newFixedThreadPool(5);

    public Scheduler() {
        this.urlManager = new UrlManager();
        this.downloader = new HttpClientDownloader();
        this.dataPipelines = new ArrayList<>();
    }

    public void schedule() throws IOException {
        int count = 0;
        String startUrl = this.spider.getStartUrl();
        this.urlManager.addUrl(startUrl);

        List<Future<Object>> futureList = new ArrayList<>();
        while (urlManager.hasUrl()) {
            String url = urlManager.getUrl();

            Future<Object> future = executorService.submit(new Task(url));
            futureList.add(future);

            if (count > size) {
                break;
            }

            count++;
        }

        try {
            for (Future<Object> future : futureList) {
                Object data = future.get();
                if (data == null) {
                    break;
                }

                for (DataPipeline dataPipeline : dataPipelines) {
                    dataPipeline.processData(data);
                }
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
    }

    public void setSize(int size) {
        this.size = size;
    }

    public void addPipeline(DataPipeline dataPipeline) {
        this.dataPipelines.add(dataPipeline);
    }

    public void setSpider(Spider spider) {
        this.spider = spider;
    }

    public void shutdown() {
        try {
            executorService.shutdown();
            executorService.awaitTermination(5, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            if (executorService.isTerminated()) {
                executorService.shutdownNow();
            }
        }
    }

    class Task implements Callable<Object> {

        private String url;

        public Task(String url) {
            this.url = url;
        }

        @Override
        public Object call() throws Exception {
            String response = getDownloader().download(url);
            Object data = getSpider().parseData(url, response);
            List<String> newUrls = getSpider().parseUrls(url, response);
            getUrlManager().addUrls(newUrls);
            return data;
        }
    }

    public UrlManager getUrlManager() {
        return urlManager;
    }

    public Downloader getDownloader() {
        return downloader;
    }


    public Spider getSpider() {
        return spider;
    }


}
